package org.jetlinks.pro.things.impl;

import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.ThingType;
import org.jetlinks.pro.things.ThingsDataRepository;
import org.jetlinks.pro.things.ThingsDataRepositorySupport;
import org.jetlinks.pro.things.data.AggregationRequest;
import org.jetlinks.pro.things.data.PropertyAggregation;
import org.jetlinks.pro.things.data.ThingPropertyDetail;
import org.jetlinks.pro.timeseries.query.AggregationData;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.function.Function3;
import reactor.function.Function4;
import reactor.function.Function5;

import javax.annotation.Nonnull;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Supplier;

public class DefaultThingsDataRepository implements ThingsDataRepository {

    private final List<ThingsDataRepositorySupport> supports = new CopyOnWriteArrayList<>();

    public DefaultThingsDataRepository() {

    }

    public DefaultThingsDataRepository(Iterable<ThingsDataRepositorySupport> supports) {
        supports.forEach(this::addSupport);
    }

    public void addSupport(ThingsDataRepositorySupport support) {
        supports.add(support);
    }

    @Override
    public Mono<Void> save(ThingType thingType, ThingMessage message) {
        return executeWith(thingType,
                           message,
                           ThingsDataRepositorySupport::save,
                           Mono::empty);
    }

    @Override
    public Flux<ThingPropertyDetail> queryProperty(ThingType thingType,
                                                   String thingId,
                                                   QueryParamEntity param) {
        return executeWith(thingType,
                           thingId,
                           param,
                           ThingsDataRepositorySupport::queryProperty,
                           Flux::empty);
    }

    @Override
    public Flux<ThingPropertyDetail> queryProperty(ThingType thingType,
                                                   String thingId,
                                                   String property,
                                                   QueryParamEntity param) {
        return executeWith(thingType,
                           thingId,
                           property,
                           param,
                           ThingsDataRepositorySupport::queryProperty,
                           Flux::empty);
    }

    @Override
    public Mono<PagerResult<ThingPropertyDetail>> queryPropertyPage(ThingType thingType,
                                                                    String thingId,
                                                                    String property,
                                                                    QueryParamEntity param) {
        return executeWith(thingType,
                           thingId,
                           property,
                           param,
                           ThingsDataRepositorySupport::queryPropertyPage,
                           Mono::empty);
    }

    @Override
    public Flux<AggregationData> aggregationProperties(ThingType thingType,
                                                       @Nonnull String thingId,
                                                       @Nonnull AggregationRequest request,
                                                       @Nonnull PropertyAggregation... properties) {
        return executeWith(thingType,
                           thingId,
                           request,
                           properties,
                           ThingsDataRepositorySupport::aggregationProperties,
                           Flux::empty);
    }


    private <T, ARG> T executeWith(ThingType thingType,
                                   ARG arg,
                                   Function3<ThingsDataRepositorySupport, ThingType, ARG, T> executor,
                                   Supplier<T> defaultIfNotSupported) {
        for (ThingsDataRepositorySupport support : supports) {
            if (support.isSupported(thingType)) {
                return executor.apply(support, thingType, arg);
            }
        }
        return defaultIfNotSupported.get();
    }

    private <T, ARG, ARG2, ARG3> T executeWith(ThingType thingType,
                                               ARG arg,
                                               ARG2 arg2,
                                               ARG3 arg3,
                                               Function5<ThingsDataRepositorySupport, ThingType, ARG, ARG2, ARG3, T> executor,
                                               Supplier<T> defaultIfNotSupported) {
        for (ThingsDataRepositorySupport support : supports) {
            if (support.isSupported(thingType)) {
                return executor.apply(support, thingType, arg, arg2, arg3);
            }
        }
        return defaultIfNotSupported.get();
    }

    private <T, ARG, ARG2> T executeWith(ThingType thingType,
                                         ARG arg,
                                         ARG2 arg2,
                                         Function4<ThingsDataRepositorySupport, ThingType, ARG, ARG2, T> executor,
                                         Supplier<T> defaultIfNotSupported) {
        for (ThingsDataRepositorySupport support : supports) {
            if (support.isSupported(thingType)) {
                return executor.apply(support, thingType, arg, arg2);
            }
        }
        return defaultIfNotSupported.get();
    }
}
